package com.hbase.zhou.telecom;

import com.hbase.zhou.db.HBaseUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.*;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Description: HBase各种过滤器汇总
 * @Author: ZhOu
 * @Date: 2018/3/20
 */
public class FilterTest {

    private void scanData(Filter filter) {
        Scan scan = new Scan();
        scan.setFilter(filter);
        scan.setReversed(true);
        try {
            Table table = HBaseUtil.getConnection().getTable(TableName.valueOf("tel_db"));
            ResultScanner results = table.getScanner(scan);
            int count = 0;
            for (Result result : results) {
                count++;
                for (Cell cell : result.rawCells()) {
                    System.out.println("rowkey=" + new String(CellUtil.cloneRow(cell))
                            + "\t family=" + new String(CellUtil.cloneFamily(cell))
                            + "\t qualifier=" + new String(CellUtil.cloneQualifier(cell))
                            + "\t value=" + new String(CellUtil.cloneValue(cell))
                            + "\t timestamp=" + cell.getTimestamp()
                    );
                }
            }
            System.out.println("\n返回结果：" + count);
            results.close();
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * RowFilter: 筛选出所有匹配行键的行
     */
    @Test
    public void rowFilterTest() {
        BinaryComparator binaryComparator = new BinaryComparator("18618316255_20170104215735".getBytes());
        Filter filter = new RowFilter(CompareFilter.CompareOp.LESS, binaryComparator);
        scanData(filter);
    }

    /**
     * PrefixFilter: 筛选行键中带有特定前缀的行，等于RowFilter + RegexComparator结合实现
     */
    @Test
    public void prefixFilterTest() {
        PrefixFilter filter = new PrefixFilter("18618316255_".getBytes());
        scanData(filter);
    }

    /**
     * KeyOnlyFilter: 返回所有行的行键，value全部为空。减少数据的传递量，可以及提升效率
     */
    @Test
    public void keyOnlyFilter() {
        KeyOnlyFilter filter = new KeyOnlyFilter();
        scanData(filter);
    }

    /**
     * RandomRowFilter:按机率随机返回结果集，返回结果集个数约等于总记录数*随机数
     * 随机数<=0,不返回结果；随机数>=1时，返回所有结果。
     */
    @Test
    public void randomRowFilter() {
        RandomRowFilter filter = new RandomRowFilter(0.01f);
        scanData(filter);
    }

    /**
     * InclusiveStopFilter: 包含上限在结果集中，不加此过滤器时搜索结果默认是前闭后开区间
     */
    @Test
    public void inclusiveFilter() {
        InclusiveStopFilter filter = new InclusiveStopFilter("18618316255_20170104215735".getBytes());
        scanData(filter);
    }

    /**
     * FirstKeyOnlyFilter: 只返回每行的第一列数据，可以提升效率
     */
    @Test
    public void firstKeyOnlyTest() {
        FirstKeyOnlyFilter filter = new FirstKeyOnlyFilter();
        scanData(filter);
    }


    /**
     * ColumnPrefixFilter: 根据列名前缀限制返回的列
     */
    @Test
    public void columnPrefixFilterTest() {
        //只获取以d开头的列
        ColumnPrefixFilter filter = new ColumnPrefixFilter("d".getBytes());
        scanData(filter);
    }


    /**
     * ColumnCountGetFilter: 每行最多返回指定个数的列，超过这个数之后就会停止扫描
     */
    @Test
    public void columnCountGetFilter() {
        //每行有4列，设置为3，将只返回一行的3列；设置为4，将返回所有的结果
        ColumnCountGetFilter filter = new ColumnCountGetFilter(3);
        scanData(filter);
    }


    /**
     * SingleColumnValueFilter: 根据某一列的值做筛选过滤
     */
    @Test
    public void singleColumnValueFilter() {
        //查询通话时长为0的记录
        SingleColumnValueFilter filter = new SingleColumnValueFilter(
                "cf1".getBytes(),
                "duration".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                "0".getBytes());
        //如果列不存，设置为true，默认是false
//        filter.setFilterIfMissing(true);
        scanData(filter);
    }


    /**
     * SingleColumnValueExcludeFilter: 和SingleColumnValueFilter用法一样，只是返回结果不包含条件列
     */
    @Test
    public void singleColumnValueExcludeTest() {
        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter(
                "cf1".getBytes(),
                "duration".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                "0".getBytes());
        scanData(filter);
    }


    /**
     * ValueFilter: 根据具体的值来筛选单元格，把一行中不能满足条件的单元格过滤掉。
     */
    @Test
    public void valueFilterTest() {
        //过滤各个字段的值中，只返回包含“80”这个值的字段
        SubstringComparator substringComparator = new SubstringComparator("80");
        ValueFilter filter = new ValueFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
        scanData(filter);
    }

    /**
     * SkipFilter: 和ValueFilter配合使用，如果行中某一列不能满足条件，这一行将会被过滤掉
     */
    @Test
    public void skipFilterTest() {
        ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator("0".getBytes()));
        SkipFilter filter = new SkipFilter(valueFilter);
        scanData(filter);
    }


    /**
     * WhileMatchFilter: 遇到不符合指定的条件数据时，终止扫描
     */
    @Test
    public void whileMatchFilter() {
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueExcludeFilter(
                "cf1".getBytes(),
                "type".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                "1".getBytes());

        WhileMatchFilter filter = new WhileMatchFilter(singleColumnValueFilter);
        scanData(filter);
    }


    /**
     * FilterList: 综合使用多个Filter，各个过滤器可以是and(MUST_PASS_ALL)和or(MUST_PASS_ONE)的关系
     */
    @Test
    public void filterListTest() {
        List<Filter> filters = new ArrayList<>();
        //搜索手机号以1863开头的，主动呼叫，且通话时长大于0的的记录
        PrefixFilter prefixFilter = new PrefixFilter("1863".getBytes());
        filters.add(prefixFilter);

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "cf1".getBytes(),
                "duration".getBytes(),
                CompareFilter.CompareOp.GREATER,
                "0".getBytes());
        filters.add(singleColumnValueFilter);

        SingleColumnValueFilter singleColumnValueFilter1 = new SingleColumnValueFilter(
                "cf1".getBytes(),
                "type".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                "0".getBytes());
        filters.add(singleColumnValueFilter1);

        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters);
        scanData(filterList);
    }
}
